package com.example.mq2.mqserver.core;

import com.example.mq2.common.MqException;
import com.example.mq2.mqserver.VirtualHost;
import com.example.mq2.mqserver.datacenter.MemoryDataCenter;

import java.io.IOException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class ConsumerManager {

    private VirtualHost virtualHost = null;
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    private Thread scannerThread = null;
    private ExecutorService consumePool = Executors.newFixedThreadPool(4);

    public ConsumerManager(VirtualHost virtualHost) {
        this.virtualHost = virtualHost;

        //创建一个扫描线程，取出对应队列名，从该队列中取出一个消费者执行一个任务
        scannerThread = new Thread(() -> {
            while(true) {
                try {
                    String queueName = tokenQueue.take();
                    MSGQueue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);
                    if(queue == null) {
                        throw new MqException("[ConsumerManager] 取到令牌后，发现该队列不存在！queueName=" + queueName);
                    }
                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        //设置守护线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    /**
     * 消费消息
     * @param queue
     */
    private void consumeMessage(MSGQueue queue) throws MqException {
        //1.从该队列中取出一个消费者
        ConsumerEnv consumerEnv = queue.chooseConsumer();
        if(consumerEnv == null) {
            System.out.println("[ConsumerManager] 消费消息时，发现该队列还没有消费者订阅~ queueName=" + queue.getName());
            return;
        }
        //2.取出一个对应该队列的消息
        Message message = virtualHost.getMemoryDataCenter().pollMessage(queue);
        if(message == null) {
            System.out.println("[ConsumerManager] 消费消息时，发现该队列还没有消息进行消费！queueName=" + queue.getName());
            return;
        }
        //3.使用线程池进行消费
        consumePool.submit(() -> {
            //先把消息放入待确认队列
            virtualHost.getMemoryDataCenter().addWaitAckMessage(queue.getName(), message);
            //开始消费
            try {
                consumerEnv.getConsumer().handlerDeliver(consumerEnv.getConsumerTag(), message.getBasicProperties(), message.getBody());
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
            if(consumerEnv.isAutoAck()) {
                if(message.getDeliverMode() == 2) {
                    try {
                        virtualHost.getDiskDataCenter().deleteMessage(queue, message);
                    } catch (IOException | MqException e) {
                        throw new RuntimeException(e);
                    }
                }
                //自动应答(直接删除消息即可)
                virtualHost.getMemoryDataCenter().removeMessage(message.getMessageId());
                virtualHost.getMemoryDataCenter().removeWaitAckMessage(queue.getName(), message.getMessageId());
            }
        });
        System.out.println("[ConsumerManager] 消费者消费消息完成！queueName=" + queue.getName() +
                ", consumerTag=" + consumerEnv.getConsumerTag());
    }

    /**
     * 将消费者添加到对应队列中
     * 每个队列会提供一个 List，用来存放所有消费者
     * @param consumerEnv
     */
    public void addConsumer(ConsumerEnv consumerEnv) throws MqException, InterruptedException {
        MSGQueue queue = virtualHost.getMemoryDataCenter().getQueue(consumerEnv.getQueueName());
        if(queue == null) {
            throw new MqException("[ConsumerManager] 队列不存在！");
        }
        queue.addConsumer(consumerEnv);
        //如果此时有消息，就要立即消费掉
        consumeMessage(queue);
    }

    /**
     * 这里通知消费者消费的逻辑就是向阻塞队列中添加该队列的名字即可
     * @param queue
     */
    public void notifyConsumer(MSGQueue queue) throws InterruptedException {
        tokenQueue.put(queue.getName());
    }
}
